package cn.doitedu.day04

import java.sql.{DriverManager, PreparedStatement}

import cn.doitedu.day01.utils.SparkUtil
import cn.doitedu.day03.beans.Student
import org.apache.spark.rdd.RDD

/**
 * @Date 22.3.31
 * @Created by HANGGE
 * @Description
 */
object C04_转换算子_MapPartitions {
  def main(args: Array[String]): Unit = {

    // 获取环境
    val sc = SparkUtil.getSc
    // 加载文件
    val rdd1 = sc.textFile("data/students.csv")

    // 2) 将每行数据切割 处理   封装在Student类中
    val res1: RDD[Student] = rdd1.map(line=>{
      val arr = line.split(",")
      //7,七娃,15,M,99,doit30
      Student(arr(0).toInt, arr(1), arr(2).toInt, arr(3), arr(4).toDouble, arr(5))
    })

    /*res1.map(stu=>stu.name)*/
  // 每个分区执行一次函数  处理的一个分区的数据  iters=>iters
    // 2个分区
 res1.foreachPartition(iters => {
   // 每个分区获取一次数据库连接
    val conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/doit30?characterEncoding=UTF-8", "root", "root")
    val ps: PreparedStatement = conn.prepareStatement("insert  into  tb_student values (?,?,?,?,?,?)")
   // 迭代器不会执行
    iters.foreach(stu => {
      ps.setInt(1, stu.sid)
      ps.setString(2, stu.name)
      ps.setInt(3, stu.age)
      ps.setString(4, stu.gender)
      ps.setDouble(5, stu.score)
      ps.setString(6, stu.grade)
      // 2) 执行SQL语句
      ps.executeUpdate()
    })
  })




  }

}
